package groupOne.app.DWD.db;

import groupOne.app.BaseAppSql;
import groupOne.common.Constant;
import groupOne.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_TradeSourceOrderXiang extends BaseAppSql {
    public static void main(String[] args) {
        new Dwd_TradeSourceOrderXiang().init(
                6009,
                2,
                "Dwd_TradeSourceOrderXiang"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"Dwd_TradeSourceOrderXiang");
        //筛选order_info表
        Table orderInfo = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['origin_amount'] origin_amount, " +
                "data['coupon_reduce'] coupon_reduce, " +
                "data['final_amount'] final_amount, " +
                "data['order_status'] order_status, " +
                "data['out_trade_no'] out_trade_no, " +
                "data['trade_body'] trade_body, " +
                "data['session_id'] session_id, " +
                "data['province_id'] province_id, " +
                "data['create_time'] create_time, " +
                "data['expire_time'] expire_time, " +
                "data['update_time'] update_time," +
                "ts " +
                "from ods_db " +
                "where `database` = 'edu' " +
                "and `table` = 'order_info' " +
                "and `type` = 'insert' ");
        tEnv.createTemporaryView("order_info",orderInfo);
        //连接page主题
        tEnv.executeSql("create table dwd_traffic_page(" +
                "common map<string,string>," +
                "data map<string,string>," +
                "ts bigint " +
                ")"+ SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRAFFIC_PAGE,"Dwd_TradeSourceOrderXiang")
        );
        //取出日志数据
        Table source = tEnv.sqlQuery("select " +
                "common['sid'] sid," +
                "common['sc'] sc " +
                "from dwd_traffic_page");
        tEnv.createTemporaryView("source",source);
        
        //join 得到sc:来源id
        Table result = tEnv.sqlQuery("select " +
                "oi.id, " +
                "oi.user_id, " +
                "oi.origin_amount, " +
                "oi.coupon_reduce, " +
                "oi.final_amount, " +
                "oi.order_status, " +
                "oi.out_trade_no, " +
                "oi.trade_body, " +
                "oi.session_id, " +
                "oi.province_id, " +
                "oi.create_time, " +
                "oi.expire_time, " +
                "oi.update_time," +
                "oi.ts," +
                "s.sc " +
                "from order_info oi join source s on oi.session_id=s.sid");
        
        tEnv.executeSql("create table dwd_source_order(" +
                "id  string," +
                "user_id  string," +
                "origin_amount  string," +
                "coupon_reduce  string," +
                "final_amount  string," +
                "order_status  string," +
                "out_trade_no  string," +
                "trade_body  string," +
                "session_id  string," +
                "province_id  string," +
                "create_time  string," +
                "expire_time  string," +
                "update_time string," +
                " ts bigint," +
                " sc string " +
                ")"+SQLUtil.getKafkaSink(Constant.TOPIC_DWD_SOURCE_ORDER));

        result.executeInsert("dwd_source_order");
    }
}
